package com.xy.biz.mc.service.impl;

import lombok.extern.slf4j.Slf4j;

import com.xy.biz.mc.domain.bo.UserConnectionBO;
import com.xy.biz.mc.service.UserConnectionService;
import com.xy.biz.mc.service.ConnectionService;
import com.xy.biz.mc.service.ServerEndPointService;
import com.xy.biz.mc.connection.SocketConnection;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

//import static newdev.biz.ws.domain.consts.MQConstant.TAG_CLOSE_CONNECTION;
//import static newdev.biz.ws.domain.consts.MQConstant.TOPIC_WS_PREFIX;

@Slf4j
@Service
public class ConnectionServiceImpl implements ConnectionService {

    @Autowired
    private ServerEndPointService serverEndPointService;
    @Autowired
    private UserConnectionService userConnectionService;
//    @Autowired
//    private RocketMQTemplate rocketMQTemplate;

    private final ConcurrentMap<Long, SocketConnection> connectionMap = new ConcurrentHashMap<>();

    @Override
    public void addConnection(Long userId, Long userGroupId, SocketConnection connection) {
        Long localServerId = serverEndPointService.getServerId();
        Long registerServerId = userConnectionService.getServerId(userId);
        if(registerServerId != null && !Objects.equals(localServerId, registerServerId)) {
            //如果之前的registerServerId未注销，且与当前用户连接上的新的localServerId不一致，
            //说明之前的连接位于其它服务实例上，则通过消息队列发消息来主动关闭之前的连接。
//            rocketMQTemplate.convertAndSend(
//                    TOPIC_WS_PREFIX + registerServerId + ":" + TAG_CLOSE_CONNECTION, userId);
        }
        SocketConnection localOldConnection = connectionMap.remove(userId);
        if(localOldConnection != null) {
            //如果用户之前连接的就位于当前服务实例上，且仍存活，则直接主动关闭之前的连接
            localOldConnection.manualClose(null);
        }

        //在当前服务实例上缓存新的连接
        connectionMap.put(userId, connection);
        //注册用户与server的路由
        userConnectionService.register(UserConnectionBO.builder().build().builder()
                .serverId(localServerId)
                .userId(userId)
                .userGroupId(userGroupId)
                .build());
    }

    @Override
    public void removeConnection(Long userId, Long userGroupId, SocketConnection connection) {
        if(null == userId) {
            return;
        }
        SocketConnection localOldConnection = connectionMap.get(userId);
        //作比较，防止误删
        if(localOldConnection != null && localOldConnection.equals(connection)) {
            connectionMap.remove(userId);
            //注销用户与server的路由
            userConnectionService.unregister(UserConnectionBO.builder()
                    .userId(userId)
                    .userGroupId(userGroupId)
                    .build());
        }
    }

    @Override
    public SocketConnection getConnection(Long userId) {
        return connectionMap.get(userId);
    }

    @Override
    public void purgeDeadConnection() {
        for(Map.Entry<Long, SocketConnection> entry : connectionMap.entrySet()) {
            SocketConnection socketConnection = entry.getValue();
            if(socketConnection.isTimeOut()) {
                socketConnection.manualClose("timeout");
//                connectionMap.remove(entry.getKey());
            }
        }
    }

}
